import java.io.File;
import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import com.microsoft.azure.storage.CloudStorageAccount;
import com.microsoft.azure.storage.blob.CloudBlobClient;
import com.microsoft.azure.storage.blob.CloudBlobContainer;
import com.microsoft.azure.storage.blob.CloudBlockBlob;
        
public class WordCount {
        
 public static class Map extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();
        
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        StringTokenizer tokenizer = new StringTokenizer(line);
        while (tokenizer.hasMoreTokens()) {
            word.set(tokenizer.nextToken());
            context.write(word, one);
        }
    }
 } 
        
 public static class Reduce extends Reducer<Text, IntWritable, Text, IntWritable> {

    public void reduce(Text key, Iterable<IntWritable> values, Context context) 
      throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        context.write(key, new IntWritable(sum));
    }
 }
        
 public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
        
    Job job = new Job(conf, "wordcount");
    job.setJarByClass(WordCount.class);
//	String[] allArgs = new GenericOptionsParser(conf, args)
//	.getRemainingArgs();
	int sizeArg = args.length;
	String[] otherArgs = {args[sizeArg - 2], args[sizeArg - 1]};
	
//	for (String p : otherArgs) {
//		Path file = new Path(StringUtils.unEscapeString(p));
//		FileSystem fs = file.getFileSystem(conf);
//		
//		System.err.println("home - " + fs.getHomeDirectory());
//		System.err.println("URI - " + fs.getUri());
//		System.err.println("work - " + fs.getWorkingDirectory());
//		
//		System.err.println("name - " + file.toUri());
//		System.err.println("Atual - " + System.getProperty("user.dir"));
//		
//	}
	
//	Path file = new Path(StringUtils.unEscapeString(otherArgs[0]));
//	FileSystem fs = file.getFileSystem(conf);
//  BufferedReader in = new BufferedReader(new FileReader("c:/apps/temp/hdfs/nm-local-dir/usercache/admin/appcache/application_1415363412012_0114/container_1415363412012_0114_01_000002/SDProject/Input/arq1.txt"));
//      String str;
//      while (in.ready()) {
//          str = in.readLine();
//          System.err.println(">>>>>>>>>>>>>>>AQUI ESTOU AQUI: " + str);
//          break;
//      }
//      in.close();
	
//    System.err.println("def_name: " + conf.get("fs.default.name"));
//    System.err.println("hadoop dir: " + conf.get("hadoop.tmp.dir"));
//    System.err.println("system dir: " + conf.get("mapred.system.dir"));
//    System.err.println("local dir: " + conf.get("mapred.local.dir"));
    
    final String storageConnectionString =
            "DefaultEndpointsProtocol=http;"
            + "AccountName=testematcher;"
            + "AccountKey=TqIn51pzE5r8CyAv0QeTTBqtd7LbS7bksSv/C8CMXAzvabqjtobID4qk5rw9yiY55p4mx5pwUC0/I1l7b19Zqw==";
    
    CloudStorageAccount account = CloudStorageAccount.parse(storageConnectionString);
    CloudBlobClient serviceClient = account.createCloudBlobClient();

    // Container name must be lower case.
    CloudBlobContainer container = serviceClient.getContainerReference("matchercluster");
    container.createIfNotExists();
    CloudBlockBlob obj = container.getBlockBlobReference("arq1.txt");
    System.err.println("CONSEGUIIIIII: " + obj.getUri().toString());
    
    
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
        
    job.setMapperClass(Map.class);
    job.setReducerClass(Reduce.class);
        
    job.setInputFormatClass(TextInputFormat.class);
    job.setOutputFormatClass(TextOutputFormat.class);
        
    FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
    FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
        
    job.waitForCompletion(true);
 }
        
}
